package com.parsechina.hpush.broker.internal.listener;

import com.parsechina.hpush.broker.internal.listener.*;
import com.parsechina.hpush.broker.internal.message.*;
import com.parsechina.hpush.broker.internal.subscriptions.Subscription;
import com.parsechina.hpush.broker.parser.ConnectMessage;
import com.parsechina.hpush.broker.parser.PublishMessage;
import com.parsechina.hpush.broker.spi.IMessaging;
import com.parsechina.hpush.broker.spi.ProtocolProcessor;
import io.netty.buffer.ByteBuf;
import org.redisson.api.RTopicAsync;
import org.redisson.api.RedissonClient;

/**
 * @author linfeng
 * @date 2019-08-20
 **/
public class ClusterMessaging implements IMessaging {

    private final RTopicAsync topicAsync;
    private final String instanceId;

    public ClusterMessaging(RedissonClient redissonClient, ProtocolProcessor processor, String instanceId) {

        this.topicAsync = redissonClient.getTopic(MessageTopic.CLUSTER_MESSAGE_TOPIC);
        this.instanceId = instanceId;
        this.topicAsync.addListenerAsync(NotifyConnectMessage.class, new NotifyConnectMessageListener(processor, instanceId));
        this.topicAsync.addListenerAsync(NotifyDisconnectMessage.class, new NotifyDisconnectMessageListener(processor, instanceId));
        this.topicAsync.addListenerAsync(NotifyPublishMessage.class, new NotifyPublishMessageListener(processor, instanceId));
        this.topicAsync.addListenerAsync(NotifySubscribeMessage.class, new NotifySubscribeMessageListener(processor, instanceId));
        this.topicAsync.addListenerAsync(NotifyUnsubscribeMessage.class, new NotifyUnsubscribeMessageListener(processor, instanceId));

    }

    @Override
    public void notifyClientConnected(ConnectMessage msg) {

        NotifyConnectMessage notifyConnectMessage = new NotifyConnectMessage(msg);
        notifyConnectMessage.setUuid(instanceId);
        topicAsync.publishAsync(notifyConnectMessage);
    }

    @Override
    public void notifyClientDisconnected(String clientId, String username) {

    }

    @Override
    public void notifyClientConnectionLost(String clientId, String username) {

    }

    @Override
    public void notifyTopicPublished(PublishMessage msg, String clientId, String username) {

        NotifyPublishMessage publishMessage = new NotifyPublishMessage(clientId, username, msg);
        publishMessage.setUuid(instanceId);
        topicAsync.publishAsync(publishMessage);
    }

    @Override
    public void notifyTopicSubscribed(Subscription sub, String username) {

    }

    @Override
    public void notifyTopicUnsubscribed(String topic, String clientId, String username) {

    }

    @Override
    public void notifyMessageAcknowledged(NotifyAcknowledgedMessage msg) {

        msg.setUuid(instanceId);
        topicAsync.publishAsync(msg);
    }

    @Override
    public void notifyIOSPush(ByteBuf msg, String clientId) {

        NotifyIOSPublishMessage publishMessage = new NotifyIOSPublishMessage(clientId, "", msg);
        publishMessage.setUuid(instanceId);
        topicAsync.publishAsync(publishMessage);
    }
}
